package com.atbeijing.D02;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Calendar;
import java.util.Random;

/**
 * 模拟数据源,温度传感器,有10个,监控是个区域
 * 泛型为流中的元素类型
 */
public class SensorSource extends RichParallelSourceFunction<SensorReading> {
    private Boolean running =true;

    //生产数据,每条数据相当于一个温度监控器产生的数据,相同的温度监控器的id一样,这里相当于有10个温度监控器
    @Override
    public void run(SourceContext sourceContext) throws Exception {
        Random random = new Random();
        String[] sensorIds = new String[10];
        double[] curFTemps = new double[10];
        for (int i = 0; i < 10; i++) {
            sensorIds[i]="sensor_"+i;
            curFTemps[i]=65+random.nextGaussian()*20;
        }

        while (running){
            long curTs = Calendar.getInstance().getTimeInMillis();
            for (int i = 0; i < 10; i++) {
                curFTemps[i] += random.nextGaussian() * 0.5;
                // 使用了上下文的collect方法来向下游发出数据
                sourceContext.collect(new SensorReading(sensorIds[i], curFTemps[i], curTs));
            }
            Thread.sleep(1000L); // 每隔1s发送一次传感器数据
        }
    }

    //取消任务
    @Override
    public void cancel() {
        running = false;
    }
}
